package com.hsogoo.avatar.registry;

import java.util.Collections;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;

import org.apache.commons.collections.CollectionUtils;

import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import com.hsogoo.avatar.NamedThreadFactory;
import com.hsogoo.avatar.utils.ExceptionUtil;

import lombok.extern.slf4j.Slf4j;

/**
 * Created by za-huangsaigang on 2018/10/22.
 */
@Slf4j
public abstract class AbstractAvatarRegistry implements AvatarRegistry {

    private final LinkedBlockingQueue<RegisterMeta> queue = new LinkedBlockingQueue<>();
    private final ExecutorService registerExecutor =  Executors.newSingleThreadExecutor(new NamedThreadFactory("register.executor", true));
    private final ScheduledExecutorService registerScheduledExecutor = Executors.newSingleThreadScheduledExecutor(new NamedThreadFactory("register.schedule.executor", true));
    private final ExecutorService localRegisterWatchExecutor = Executors.newSingleThreadExecutor(new NamedThreadFactory("local.register.watch.executor", true));
    //local cache
    private static Map<ServiceMeta, Set<RegisterMeta>> providerContainer = Maps.newConcurrentMap();
    // Consumer已订阅的信息
    private final AtomicReference<Set<ServiceMeta>> subscribeSet = new AtomicReference<>(Sets.newConcurrentHashSet());
    // Provider已发布的注册信息
    private final ConcurrentMap<RegisterMeta, RegisterState> registerMetaMap = Maps.newConcurrentMap();

    private final AtomicBoolean shutdown = new AtomicBoolean(false);

    public AbstractAvatarRegistry() {
        registerExecutor.execute(new Runnable() {

            @Override
            public void run() {
                while (!shutdown.get()) {
                    RegisterMeta meta = null;
                    try {
                        meta = queue.take();
                        //预注册
                        registerMetaMap.put(meta, RegisterState.PREPARE);
                        doRegister(meta);
                    } catch (InterruptedException e) {
                        log.warn("[register.executor] interrupted.");
                    } catch (Throwable t) {
                        if (meta != null) {
                            log.error("Register [{}] fail: {}, will try again...", meta.getServiceMeta(), ExceptionUtil.getDetailMessage(t));

                            // 间隔一段时间再重新入队, 让出cpu
                            final RegisterMeta finalMeta = meta;
                            registerScheduledExecutor.schedule(new Runnable() {

                                @Override
                                public void run() {
                                    queue.add(finalMeta);
                                }
                            }, 1, TimeUnit.SECONDS);
                        }
                    }
                }
            }
        });

        localRegisterWatchExecutor.execute(new Runnable() {

            @Override
            public void run() {
                while (!shutdown.get()) {
                    try {
                        Thread.sleep(3000);
                        doCheckRegisterNodeStatus();
                    } catch (InterruptedException e) {
                        log.warn("[local.register.watch.executor] interrupted.");
                    } catch (Throwable t) {
                        if (log.isWarnEnabled()) {
                            log.warn("Check register node status fail: {}, will try again...", ExceptionUtil.getDetailMessage(t));
                        }
                    }
                }
            }
        });
    }

    @Override
    public void register(RegisterMeta meta) {
        queue.add(meta);
    }

    @Override
    public void subscribe(ServiceMeta serviceMeta) {
        subscribeSet.get().add(serviceMeta);
        doSubscribe(serviceMeta);
    }

    @Override
    public Set<RegisterMeta> lookup(ServiceMeta serviceMeta) {
        Set<RegisterMeta> registerMetas = providerContainer.get(serviceMeta);

        if (CollectionUtils.isEmpty(registerMetas)) {
            return Collections.emptySet();
        }
        return registerMetas;
    }

    /**
     * 服务节点增加
     * @param registerMeta
     */
    protected void providerAdded(RegisterMeta registerMeta){
        ServiceMeta serviceMeta = registerMeta.getServiceMeta();
        Set<RegisterMeta> providers = providerContainer.get(serviceMeta);
        if(CollectionUtils.isEmpty(providers)){
            providers = Sets.newConcurrentHashSet();
        }
        providers.add(registerMeta);
        providerContainer.put(serviceMeta, providers);
    }

    protected void providerRemoved(RegisterMeta registerMeta){
        ServiceMeta serviceMeta = registerMeta.getServiceMeta();
        Set<RegisterMeta> providers = providerContainer.get(serviceMeta);
        if(CollectionUtils.isNotEmpty(providers)){
            providers.remove(registerMeta);
        }
        if(providers.isEmpty()){
            providerContainer.remove(serviceMeta);
        }
    }

    protected abstract void doRegister(RegisterMeta meta);

    protected abstract void doCheckRegisterNodeStatus();

    protected abstract void doSubscribe(ServiceMeta serviceMeta);

    protected ConcurrentMap<RegisterMeta, RegisterState> getRegisterMetaMap(){
        return registerMetaMap;
    }

}
